package cn.itcast.function;

import cn.itcast.bean.IndexBean;

import com.alibaba.fastjson.JSON;
import org.apache.flink.streaming.api.functions.windowing.RichAllWindowFunction;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;
import org.apache.hadoop.hbase.client.Put;
import scala.collection.Iterable;

import java.util.ArrayList;
import java.util.List;

public class IndexPutHbaseWindowFunction extends RichAllWindowFunction<IndexBean, List<Put>, TimeWindow> {


    @Override
    public void apply(TimeWindow window, java.lang.Iterable<IndexBean> values, Collector<List<Put>> out) throws Exception {
        //         * 1.新建list
        List<Put> list = new ArrayList<>();
//         * 2.轮询窗口集合数据
        for (IndexBean value : values) {
//         * 3.封装put对象
            String rowKey=value.getIndexCode()+value.getTradeTime();
            Put put = new Put(rowKey.getBytes());
//         * 4.bean转字符串
            String toJSONString = JSON.toJSONString(value);
//         * 5.并封装集合数据
            put.addColumn("info".getBytes(),"data".getBytes(),toJSONString.getBytes());
            list.add(put);
        }
    }
}
